package com.itheima.article.stream;

import com.alibaba.fastjson.JSON;
import com.itheima.article.dto.ArticleVisitStreamMess;
import com.itheima.behaviour.dto.UpdateArticleMess;
import com.itheima.common.constants.BusinessConstants;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.Aggregator;
import org.apache.kafka.streams.kstream.Initializer;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @version 1.0
 * @description 文章实时计分流处理器
 * @package com.itheima.article.stream
 */
@Slf4j
@Configuration
public class HotArticleStreamHandler {

    @Bean
    public KStream<String,String> kStream(StreamsBuilder streamsBuilder){
        //1.监听输入主题
        KStream<String, String> stream = streamsBuilder.stream(BusinessConstants.MqConstants.HOT_ARTICLE_SCORE_TOPIC);
        //2.流中数据处理, 在行为微服点赞时，发送消息时没有指定key
        // stream里的元素 kafka消息 UpdateArticleMess
        // 初始化
        stream
            //3.获取key 文章id
            .map((key,value)->{
            // key=>null
            //value = UpdateArticleMess 消息内容 {type:VIEWS, articleId:101, num:1}
            UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
            Long articleId = mess.getArticleId();
            //【注意】Kafka的key与value都是String类型
            return new KeyValue<>(articleId.toString(), value);
        })
        //4.按文章id分组
        .groupBy((articleId,msgJson)->articleId)
        //5.统计一段时间内容的消息
        .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
        //6.聚合统计, 流中元素 key=文章id, value=消息的json字符串(UpdateArticleMess)
        .aggregate(() -> {
            // 初始化最终要的聚合的结果对象
            ArticleVisitStreamMess result = new ArticleVisitStreamMess();
            String json = JSON.toJSONString(result);
            return json;
        }, new Aggregator<String, String, String>() {
            @Override
            // 循环被调用, 累加消息
            public String apply(String key, String value, String aggregate) {
                // key=文章id
                //long articleId = Long.parseLong(key);
                // value=消息对象的json字符串 UpdateArticleMess
                UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
                // aggregate 聚合结果  ArticleVisitStreamMess json字符串
                ArticleVisitStreamMess result = JSON.parseObject(aggregate, ArticleVisitStreamMess.class);
                // 文章id
                result.setArticleId(mess.getArticleId());
                int type = mess.getType().getValue();
                switch (type){
                    // 收藏
                    case 1: result.setCollect(result.getCollect() + mess.getNum());break;
                    case 2: result.setComment(result.getComment() + mess.getNum());break;
                    case 3: result.setLike(result.getLike() + mess.getNum());break;
                    case 4: result.setView(result.getView() + mess.getNum());break;
                }
                String resultJson = JSON.toJSONString(result);
                System.err.println(resultJson);
                return resultJson;
            }
        },  // 取个别名，唯一即可
            Materialized.as("hotArticleAggResult"))
        //7.转stream
        .toStream()
        //8.转kafka消息格式
        .map((key,value)->new KeyValue<>(key.key().toString(), value))
        //9.输出主题
        .to(BusinessConstants.MqConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);
        //10.返回stream
        return stream;
    }

    public static void main(String[] args) {
        List<UpdateArticleMess> list = new ArrayList<>();
        // 8个阅读消息
        for (int i = 0; i < 8; i++) {
            UpdateArticleMess msg = new UpdateArticleMess();
            msg.setType(UpdateArticleMess.UpdateArticleType.VIEWS);
            msg.setNum(1);
            msg.setArticleId(101l);
            list.add(msg);
        }
        // 5 个点赞
        for (int i = 0; i < 5; i++) {
            UpdateArticleMess msg = new UpdateArticleMess();
            msg.setType(UpdateArticleMess.UpdateArticleType.LIKES);
            msg.setNum(1);
            msg.setArticleId(101l);
            list.add(msg);
        }

        for (int i = 0; i < 3; i++) {
            UpdateArticleMess msg = new UpdateArticleMess();
            msg.setType(UpdateArticleMess.UpdateArticleType.COMMENT);
            msg.setNum(1);
            msg.setArticleId(101l);
            list.add(msg);
        }

        UpdateArticleMess msg = new UpdateArticleMess();
        msg.setType(UpdateArticleMess.UpdateArticleType.COLLECTION);
        msg.setNum(1);
        msg.setArticleId(101l);
        list.add(msg);

        System.out.println("共有" + list.size());

        //calc(list);
        // 不能使用java对象，只能用字符串, Kafka 指定的key与value都是字符串
        calc2(list);
        System.out.println(list);
    }


    private static void calc2(List<UpdateArticleMess> list) {
        //{"collect":1,"comment":3,"like":5,"view":8}
        ArticleVisitStreamMess result = new ArticleVisitStreamMess();
        String json = JSON.toJSONString(result);
        // 17个消息对象转成json字符串
        List<String> jsonList = list.stream().map(JSON::toJSONString).collect(Collectors.toList());
        for (String msgJson : jsonList) {
            //2次
            json = aggregate(msgJson,json);
        }
        // json={}101;
        //json = aggregate(msgJson,json);
        //json = aggregate(msgJson,json);

        System.out.println(json);
    }

    private static String aggregate(String msgJson, String json) {
        // msgJson -> 1个消息 UpdateArticleMess
        // json 聚合结果 ArticleVisitStreamMess
        ArticleVisitStreamMess result = JSON.parseObject(json, ArticleVisitStreamMess.class);
        // 消息对象
        UpdateArticleMess mess = JSON.parseObject(msgJson, UpdateArticleMess.class);
        int type = mess.getType().getValue();
        switch (type){
            // 收藏
            case 1: result.setCollect(result.getCollect() + mess.getNum());break;
            case 2: result.setComment(result.getComment() + mess.getNum());break;
            case 3: result.setLike(result.getLike() + mess.getNum());break;
            case 4: result.setView(result.getView() + mess.getNum());break;
        }
        return JSON.toJSONString(result);
    }


    private static void calc(List<UpdateArticleMess> list) {
        ArticleVisitStreamMess result = new ArticleVisitStreamMess();

        for (UpdateArticleMess mess : list) {
            result.setArticleId(mess.getArticleId());

            int type = mess.getType().getValue();
            switch (type){
                // 收藏
                case 1: result.setCollect(result.getCollect() + mess.getNum());break;
                case 2: result.setComment(result.getComment() + mess.getNum());break;
                case 3: result.setLike(result.getLike() + mess.getNum());break;
                case 4: result.setView(result.getView() + mess.getNum());break;
            }
        }
        System.out.println(JSON.toJSONString(result));
    }

}
